package com.zhoupu.mqtt.core.config;

import com.alibaba.fastjson.JSON;
import com.zhoupu.mqtt.core.customerconst.ChannelName;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHeaders;

import java.util.UUID;

/**
 * MQTT配置类
 *
 * @author Zhouyongli
 * @version 2024/09/12 11:26
 **/

@Configuration
public class MqttClientConfig {

    private final int QOS = 2;
    private final String[] TOPICS = {"first_topic", "second_topic", "third_topic", "four_topic"};

    private final MqttPropertyConfig mqttConfig;

    @Autowired
    public MqttClientConfig(MqttPropertyConfig mqttPropertyConfig){
        this.mqttConfig = mqttPropertyConfig;
    }

    @Bean
    public MqttPahoClientFactory mqttClientFactory(){

        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
        mqttConnectOptions.setUserName(mqttConfig.getUserName());
        mqttConnectOptions.setPassword(mqttConfig.getPassword().toCharArray());
        mqttConnectOptions.setServerURIs(mqttConfig.getServerUri());
        mqttConnectOptions.setKeepAliveInterval(mqttConfig.getKeepAlive());
        mqttConnectOptions.setAutomaticReconnect(mqttConfig.getAutoReconnect());

        DefaultMqttPahoClientFactory mqttPahoClientFactory = new DefaultMqttPahoClientFactory();
        mqttPahoClientFactory.setConnectionOptions(mqttConnectOptions);
        return mqttPahoClientFactory;
    }


    private String createClientId(){
        return UUID.randomUUID().toString();
    }


//    @Bean
//    public MessageChannel mqttInputChannel(){
//        return new DirectChannel();
//    }
//
//    @Bean
//    public MessageChannel mqttOutboundChannel(){
//        return new DirectChannel();
//    }
//
//
//    /**
//     * 配置 mqttClient 发布消息
//     * @return
//     */
//    @Bean
//    @ServiceActivator(inputChannel = ChannelName.MQTT_OUTBOUND_CHANNEL)
//    public MessageHandler mqttOutbound(){
//        MqttPahoMessageHandler mqttPahoMessageHandler = new MqttPahoMessageHandler(createClientId(), mqttClientFactory());
//        mqttPahoMessageHandler.setAsync(true);
//        mqttPahoMessageHandler.setDefaultQos(QOS);
//        mqttPahoMessageHandler.setDefaultRetained(false); // 不保留信息
//        return mqttPahoMessageHandler;
//    }
//
//
    /**
     *  配置 mqttClient 监听消息
     * @return
     */
    @Bean
    public MessageProducer mqttInput(){
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
                createClientId(),
                mqttClientFactory(),
                TOPICS);
        adapter.setCompletionTimeout(30000L);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(QOS);
        adapter.setOutputChannelName(ChannelName.MQTT_INPUT_CHANNEL);
        return adapter;
    }
//
//
//    /**
//     * mqtt 监听的 消息处理器
//     * @return
//     */
//    @Bean
//    @ServiceActivator(inputChannel = ChannelName.MQTT_INPUT_CHANNEL)
//    public MessageHandler mqttInputHandler(){
//        return message -> {
//            MessageHeaders headers = message.getHeaders();
//
//            // String headerStr = JSON.toJSONString(headers);
//            // {"mqtt_receivedRetained":false,"mqtt_id":1,"mqtt_duplicate":false,"id":"b08ba36a-9ce9-4717-e315-ad7231d126cf","mqtt_receivedTopic":"first_topic","mqtt_receivedQos":2,"timestamp":1726125110332}
//            String topic = headers.get("mqtt_receivedTopic").toString();
//            String payload = message.getPayload().toString();
//            System.out.println("message topic: " + topic);
//            System.out.println("message payload: " + payload);
//
//        };
//    }
//



    @Bean
    public IntegrationFlow mqttOutboundFlow(){

        MqttPahoMessageHandler mqttPahoMessageHandler =
                new MqttPahoMessageHandler(createClientId(), mqttClientFactory());

        mqttPahoMessageHandler.setAsync(true);
        mqttPahoMessageHandler.setDefaultQos(QOS);
        mqttPahoMessageHandler.setDefaultRetained(false); // 不保留信息

        return IntegrationFlows
                .from(ChannelName.MQTT_OUTBOUND_CHANNEL)
                .handle(mqttPahoMessageHandler)
                .get();
    }



    @Bean
    public IntegrationFlow mqttInputFlow(){

        return IntegrationFlows
                .from(ChannelName.MQTT_INPUT_CHANNEL)
                .handle(message -> {

                    String payload = message.getPayload().toString();
                    String topic = message.getHeaders().get("mqtt_receivedTopic").toString();

                    System.out.println("mqtt-topic: " + topic);
                    System.out.println("mqtt-payload: " + payload);
                    System.out.println(JSON.toJSONString(message.getHeaders()));

                })
                .get();
    }







}
